package belf.migrate.engine.job;

import belf.migrate.api.common.JobMode;
import belf.migrate.api.job.Job;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.engine.BelfEvn;
import belf.migrate.api.taskconf.Task;
import lombok.extern.slf4j.Slf4j;

import static belf.migrate.api.job.JobCode.*;

@Slf4j
public class BaseJobRunner implements JobRunner {
    protected final Job job;
    protected final Task task;
    protected final TaskConf taskConf;
    protected JobContext jobContext;

    public BaseJobRunner(Job job) {
        this.job = job;
        this.task = job.getTask();
        this.taskConf = task.getTaskConf();
    }
    /**
     * 检查环境是否满足，并初始化执行任务需要的环境JobContext
     */
    public JobResult check() {
        JobResult jobResult = new JobResult(SUCCESS, "任务执行成功");

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建源端catalog
        AbstractJdbcCatalog SourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
        if (null == SourceCatalog) {
            log.error("No Source Catalog for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());

            jobResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSinkConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");

            return jobResult;
        }
        jobContext.setSourceCatalog(SourceCatalog);

        // 创建目标catalog
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());

            jobResult = new JobResult(NO_SINK_CATALOG, "提交Job失败，没有找到源端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return jobResult;
        }
        jobContext.setSinkCatalog(sinkCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);

        return jobResult;
    }


    @Override
    public JobResult execute() {
        return null;
    }

    @Override
    public JobMode getMode() {
        return JobMode.BATCH;
    }

    @Override
    public Job getJob() {
        return job;
    }

    public JobContext getJobContext() {
        return jobContext;
    }

    public Task getTask() {
        return task;
    }

    public TaskConf getTaskConf() {
        return taskConf;
    }
}
